package xubo.wangcaifeng.love.method

import java.util.Properties

import com.typesafe.config.ConfigFactory
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}

object Need1Bycore {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("按照地区纬度查询")
      .setMaster("local[*]")
      .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)
    val sqlc = new SQLContext(sc)
    //读取文件

    val frame: DataFrame = sqlc.read.parquet("data/parquet/")
    //注册一张表
    frame.registerTempTable("tmp")
    //使用sqlc来查询数据   case  when
    val dataFrema: DataFrame = sqlc.sql(
      """
        |select provincename,cityname,
        |          sum(case when requestmode = 1 and processnode >= 1 then 1 else 0 end) as yres,
        |          sum(case when requestmode = 1 and processnode >= 2 then 1 else 0 end) as valid,
        |          sum(case when requestmode = 1 and processnode >= 3 then 1 else 0 end) as res,
        |          sum(case when iseffective = 1 and isbilling = 1 and isbid = 1 and adorderid != 0 then 1 else 0 end) as jingjia,
        |          sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then 1 else 0 end) as succjingjia,
        |          sum(case when requestmode = 2 and iseffective = 1 then 1 else 0 end) as showse,
        |          sum(case when requestmode = 3 and iseffective = 1 then 1 else 0 end) as clickss,
        |          sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then winprice/1000 else 0.0 end) as adconsumer,
        |          sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then adpayment/1000 else 0.0 end) as adpaymoney
        |          from tmp group by provincename,cityname
      """.stripMargin)
    //读取配置文件
    val reader = ConfigFactory.load()
    val props = new Properties()
    props.setProperty("driver", reader.getString("db.default.driver"))
    props.setProperty("user", reader.getString("db.default.user"))
    props.setProperty("password", reader.getString("db.default.password"))
    dataFrema.write.mode(SaveMode.Overwrite)
      .jdbc(reader.getString("db.default.url"),"localcount1",props)
    /*val url = "jdbc:mysql://localhost:3306/dmt?characterEncoding=utf8"
    val tname = "localcount1"
    val p = new Properties()
    p.setProperty("user","root")
    p.setProperty("password","217410")
    dataFrema.write.jdbc(url,tname,p)*/
    sc.stop()

  }
}
